package com.svavo.rxjava2;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import java.util.ArrayList;
import java.util.List;
import java.util.logging.Logger;

public class RxJava2Test1 {
    private static Logger logger = Logger.getLogger("RxJava2Test1");

    public static void main(String[] args) {
        //test1();
        //test2();
        //test3();
        test6();
    }

    private static void test1(){
        Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() { //创建被观察者
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                logger.info( "=========================currentThread name: " + Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        });

        Observer observer = new Observer<Integer>() { //创建观察者
            @Override
            public void onSubscribe(Disposable d) {
                logger.info("======================onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                logger.info("======================onNext " + integer);
            }

            @Override
            public void onError(Throwable e) {
                logger.info("======================onError");
            }

            @Override
            public void onComplete() {
                logger.info("======================onComplete");
            }
        };

        observable.subscribe(observer); //订阅
    }

    private static void test2(){
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("Hello Observer");
                e.onComplete();
            }
        });

        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                logger.info("=============onNext " + s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                logger.info("=============onComplete ");
            }
        };

        observable.subscribe(observer);
    }

    private static void test3(){ //发送的事件不可以超过10个以上
        Observable.just(1, 2, 3)
                .subscribe(new Observer < Integer > () {
                    @Override
                    public void onSubscribe(Disposable d) {
                        logger.info("=================onSubscribe");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        logger.info( "=================onNext " + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        logger.info( "=================onError ");
                    }

                    @Override
                    public void onComplete() {
                        logger.info( "=================onComplete ");
                    }
                });
    }

    private static void test4(){ //可以传入多于10个的变量数组
        Integer array[] = {1, 2, 3, 4};
        Observable.fromArray(array)
                .subscribe(new Observer < Integer > () {
                    @Override
                    public void onSubscribe(Disposable d) {
                        logger.info("=================onSubscribe");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        logger.info( "=================onNext " + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        logger.info( "=================onError ");
                    }

                    @Override
                    public void onComplete() {
                        logger.info( "=================onComplete ");
                    }
                });
    }

    private static void test5(){ //发送一个 List 集合数据给观察者
        List<Integer> list = new ArrayList<>();
        list.add(0);
        list.add(1);
        list.add(2);
        list.add(3);
        Observable.fromIterable(list)
                .subscribe(new Observer < Integer > () {
                    @Override
                    public void onSubscribe(Disposable d) {
                        logger.info( "=================onSubscribe");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        logger.info( "=================onNext " + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        logger.info( "=================onError ");
                    }

                    @Override
                    public void onComplete() {
                        logger.info( "=================onComplete ");
                    }
                });
    }

    private static void test6(){ //同时发送一定范围的事件序列
        Observable.range(2, 5)
                .subscribe(new Observer < Integer > () {
                    @Override
                    public void onSubscribe(Disposable d) {
                        logger.info( "==============onSubscribe ");
                    }

                    @Override
                    public void onNext(Integer aLong) {
                        logger.info( "==============onNext " + aLong);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

}
